package com.wk.springboot.config;

import com.wk.springboot.batch.PersonItemProcessor;
import com.wk.springboot.bean.Person;
import com.wk.springboot.bean.Player;
import com.wk.springboot.listener.MyChunkListener;
import com.wk.springboot.listener.MyItemProcessingListener;
import com.wk.springboot.listener.MyItemReaderListener;
import com.wk.springboot.listener.MyItemWriterListener;
import com.wk.springboot.listener.MyStepExecutionListener;
import com.wk.springboot.taskletStep.FileDeletingTasklet;
import com.wk.springboot.taskletStep.MyTaskletStep;
import javax.sql.DataSource;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.step.tasklet.MethodInvokingTaskletAdapter;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;

/**
 * @Classname StepConfig
 * @Description step的配置
 * @Date 2020/4/7 10:57
 * @Author by weikai
 */
@Configuration
public class StepConfig {

  @Autowired
  public StepBuilderFactory stepBuilderFactory;


  /**
   * 读取sample-data.csv数据写入Person中
   */
  @Bean
  public FlatFileItemReader<Person> reader() {
    return new FlatFileItemReaderBuilder<Person>()
        .name("personItemReader")
        .resource(new ClassPathResource("sample-data.csv"))
        .delimited()
        .names(new String[]{"firstName", "lastName"})
        .fieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
          setTargetType(Person.class);
        }})
        .build();
  }

  @Bean
  public PersonItemProcessor processor() {
    return new PersonItemProcessor();
  }

  /**
   * DataSource已经被@EnableBatchProcessing通过配置文件自动创建
   */
  @Bean
  public JdbcBatchItemWriter<Person> writer(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<Person>()
        .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
        .sql("INSERT INTO people (first_name, last_name) VALUES (:firstName, :lastName)")
        .dataSource(dataSource)
        .build();
  }

  /**
   * DataSource已经被@EnableBatchProcessing通过配置文件自动创建
   */
  @Bean
  public JdbcBatchItemWriter<Person> player(DataSource dataSource) {
    return new JdbcBatchItemWriterBuilder<Person>()
        .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
        .sql("INSERT INTO people (first_name, last_name) VALUES (:firstName, :lastName)")
        .dataSource(dataSource)
        .build();
  }

  /**
   * 每个step包含reader，processor，writer chunk：定义一次要写多少数据。左右的Person分别表示ItemReader<Person>和ItemWriter<Person>
   * spring PlatformTransactionManager 可以保证开始和提交事务 processor是可选的，因为数据本身可以从reader->writer
   * startLimit(1)：表示step只能执行一次；如果尝试再次测试，会报StartLimitExceededException错误；默认值为Integer.MAX_VALUE
   * allowStartIfComplete=true，表示step可以重启
   * 增加一个ChunkListener
   */
  @Bean
  public Step step1(JdbcBatchItemWriter<Person> writer,
      ItemReader<Person> reader,
      PlatformTransactionManager transactionManager,
      MyChunkListener myChunkListener,
      MyItemReaderListener myItemReaderListener,
      MyItemProcessingListener myItemProcessingListener,
      MyItemWriterListener myItemWriterListener) {
    return stepBuilderFactory.get("step1")
        .transactionManager(transactionManager)
        .<Person, Person>chunk(5)
        .reader(reader)
        .processor(processor())
        .writer(writer)
        .listener(myItemProcessingListener)
        .listener(myItemWriterListener)
        .listener(myItemReaderListener)
        .listener(myChunkListener)
        .build();
  }

  /**
   * .faultTolerant()产生一个FaultRolerantStepBuilder
   * skipLimit()：当这个step抛出Exception这个错误，可以允许跳过10；当跳过的次数达到10后，将会报错
   * skip表示发生哪个错误后，跳过
   * noSkip表示发生这个错误时，不跳过
   * noRollback:如果跳过的步骤不会引起writer回滚的，则可以这样配置
   * @param writer
   * @param transactionManager
   * @return
   */
  @Bean
  public Step step2(JdbcBatchItemWriter<Person> writer,
      PlatformTransactionManager transactionManager) {
    return stepBuilderFactory.get("step2").transactionManager(transactionManager)
        .<Person, Person>chunk(5)
        .reader(reader())
        .processor(processor())
        .writer(writer)
        .faultTolerant()
        .skipLimit(10)
        .skip(Exception.class)
        .noSkip(NegativeArraySizeException.class)
        .noRollback(NegativeArraySizeException.class)
        .allowStartIfComplete(true)
        .build();
  }

  /**
   * step3是最后一个步骤，startLimit=2是有必要的，避免ste2一直失败一直重试
   * retryLimit=3重试次数为3
   * retry:当遇到这个问题后，重试
   * @param writer
   * @param transactionManager
   * @return
   */
  @Bean
  public Step step3(JdbcBatchItemWriter<Person> writer,
      PlatformTransactionManager transactionManager) {
    return stepBuilderFactory.get("step3").transactionManager(transactionManager)
        .<Person, Person>chunk(5)
        .reader(reader())
        .processor(processor())
        .writer(writer)
//        .startLimit(2)
        .faultTolerant()
        .retryLimit(3)
        .retry(Exception.class)
        .build();
  }

  /**
   *  配置transactionAtrribute
   * @return
   */
  @Bean
  public Step step4(JdbcBatchItemWriter<Person> writer) {
    DefaultTransactionAttribute attribute = new DefaultTransactionAttribute();
    attribute.setPropagationBehavior(Propagation.REQUIRED.value());
    attribute.setIsolationLevel(Isolation.DEFAULT.value());
    attribute.setTimeout(30);

    return this.stepBuilderFactory.get("step4")
        .<Person, Person>chunk(2)
        .reader(reader())
        .writer(writer)
        .transactionAttribute(attribute)
        .build();
  }

  /**
   *  配置一个taskletStep
   * @return
   */
  @Bean
  public Step taskletStep(MyTaskletStep myTaskletStep) {

    return this.stepBuilderFactory.get("taskletStep")
        .tasklet(myTaskletStep)
        .allowStartIfComplete(true)
        .build();
  }

  /**
   *  配置一个taskletAdapter，当其他步骤执行完成之后，要修改这个状态
   * @return
   */
  @Bean
  public Step taskletAdapter(MethodInvokingTaskletAdapter methodInvokingTaskletAdapter) {

    return this.stepBuilderFactory.get("taskletAdapter")
        .tasklet(methodInvokingTaskletAdapter)
        .allowStartIfComplete(true)
        .build();
  }

  /**
   *  配置一个fileDeletingTasklet，当其他步骤执行完成之后，删除指定目录下的文件
   * @return
   */
  @Bean
  public Step fileDeletingTaskletStep(FileDeletingTasklet fileDeletingTasklet) {

    return this.stepBuilderFactory.get("fileDeletingTaskletStep")
        .tasklet(fileDeletingTasklet)
        .build();
  }

  /**
   * 自定义一个step
   * 定义这个step是可以重启的
   * @return
   */
  @Bean
  public Step myStep(MyTaskletStep myTaskletStep, MyStepExecutionListener myStepExecutionListener){

    return this.stepBuilderFactory.get("myStep")
        .tasklet(myTaskletStep)
        .allowStartIfComplete(true)
        .listener(myStepExecutionListener)
        .build();
  }

  /**
   * flatFileStep
   * @return
   */
  @Bean
  public Step flatFileStep(FlatFileItemReader flatFileItemReader,
      MyItemReaderListener myItemReaderListener){
    return this.stepBuilderFactory.get("flatFileStep")
        .< Player,Player>chunk(5)
        .reader(flatFileItemReader)
        .listener(myItemReaderListener)
        .build();

  }
}
